package com.xtck.ha.fault.modules.service.impl;

import cn.hutool.core.date.DatePattern;
import cn.hutool.http.HttpResponse;
import cn.hutool.http.HttpUtil;
import cn.hutool.http.Method;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.metadata.OrderItem;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.xtck.ha.fault.framework.kafka.IKafkaProducer;
import com.xtck.ha.fault.framework.kafka.PushFaultMsg;
import com.xtck.ha.fault.framework.page.PageResult;
import com.xtck.ha.fault.framework.query.Query;
import com.xtck.ha.fault.framework.utils.JsonResult;
import com.xtck.ha.fault.framework.utils.MessageToJsonUtil;
import com.xtck.ha.fault.modules.convert.FaultTaskConvert;
import com.xtck.ha.fault.modules.dao.FaultTaskDAO;
import com.xtck.ha.fault.modules.dao.FmeaInfoDAO;
import com.xtck.ha.fault.modules.dao.ModInfoDAO;
import com.xtck.ha.fault.modules.pojo.dto.FaultTaskDTO;
import com.xtck.ha.fault.modules.pojo.entity.FaultTask;
import com.xtck.ha.fault.modules.pojo.entity.FmeaInfo;
import com.xtck.ha.fault.modules.pojo.entity.ModInfo;
import com.xtck.ha.fault.modules.pojo.vo.FaultTaskVO;
import com.xtck.ha.fault.modules.pojo.vo.ResultDataVo;
import com.xtck.ha.fault.modules.service.FaultTaskService;
import com.xtck.ha.fault.schedule.DynamicTaskService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
* @Description: FaultTaskService实现类
* @author liuyuan
* @date 2025-06-18
*/
@Service
@Slf4j
public class FaultTaskServiceImpl extends ServiceImpl<FaultTaskDAO, FaultTask>  implements FaultTaskService {
        @Value("${web.predict-result}")
        private String predictResultUrl;
        @Autowired
        private FaultTaskDAO faultTaskDAO;
        @Autowired
        private ModInfoDAO modInfoDAO;
        @Autowired
        FmeaInfoDAO fmeaInfoDAO;
        @Autowired
        IKafkaProducer iKafkaProducer;
        @Resource
        private DynamicTaskService dynamicTaskService;

        /**
        * 获取分页对象
        * @param query   分页参数
        */
        protected IPage<FaultTask> getPage(Query query) {
                Page<FaultTask> page = new Page<FaultTask>(query.getPage(), query.getLimit());
                // 排序
                if(StringUtils.isNotBlank(query.getOrder())){
                        if(query.isAsc()) {
                                return page.addOrder(OrderItem.asc(query.getOrder()));
                        }else {
                                return page.addOrder(OrderItem.desc(query.getOrder()));
                        }
                }
                return page;
        }
        @Override
        public PageResult<FaultTaskVO> page(FaultTaskDTO query) {
                IPage<FaultTask> page = faultTaskDAO.selectPage(getPage(query), getWrapper(query));
                return new PageResult<FaultTaskVO>(FaultTaskConvert.INSTANCE.convertList(page.getRecords()), page.getTotal());
        }

        private QueryWrapper<FaultTask> getWrapper(FaultTaskDTO query){
                QueryWrapper<FaultTask> wrapper = new QueryWrapper<>();
                if (StringUtils.isNotBlank(query.getSatelliteCode()) )
                        wrapper.eq("satellite_code",query.getSatelliteCode());
                if (StringUtils.isNotBlank(query.getTaskName()) )
                        wrapper.like("task_name",query.getTaskName());
                if (null != (query.getIsSchedule()) )
                        wrapper.eq("is_schedule",query.getIsSchedule());
                wrapper.orderByDesc("create_time");

                return wrapper;
        }

        @Override
        @Transactional(rollbackFor = Exception.class)
        public FaultTask save(FaultTaskVO vo) {
                FaultTask entity = FaultTaskConvert.INSTANCE.convert(vo);
                entity.setCreateTime(new Date());
                entity.setStatus(true);
                entity.setIsFirstExec(true);
                baseMapper.insert(entity);
                return entity;
        }

        @Override
        @Transactional(rollbackFor = Exception.class)
        public JsonResult<String> execTask(FaultTask faultTask) {
                String cron = checkCron(faultTask.getCron());
                //  定义任务逻辑
                Runnable task = () -> {
                        faultProcess(faultTask);
                };
                //  启动任务
                dynamicTaskService.startOrUpdateTask(String.valueOf(faultTask.getId()), cron, task);
                FaultTask taskForDb = faultTaskDAO.selectById(faultTask.getId());
                if (null != taskForDb && !taskForDb.getStatus()){
                        taskForDb.setStatus(true);
                        updateById(taskForDb);
                }
                return JsonResult.ok("任务启动成功: " + faultTask.getTaskName());
        }

        private void faultProcess(FaultTask faultTask){
                FaultTask taskInfo = faultTaskDAO.selectById(faultTask.getId());
                /**查询对应模型信息*/
                List<ModInfo> modInfoList = modInfoDAO.selectByModName(taskInfo.getModelName());
                if (null == modInfoList ){
                        log.error("未查询到对应模型");
                        return;
                }
                else if (modInfoList.size() != 1){
                        log.error("模型名称重复");
                        return;
                }
                else {
                        /**构建参数，调用查询数据接口*/
                        ModInfo modInfo = modInfoList.get(0);
                        ResultDataVo resultDataVo = ResultDataVo.builder()
                                .satelliteCode(taskInfo.getSatelliteCode())
                                .paramCode(taskInfo.getParamCode())
                                .build();
                        /**是否是第一次执行将会影响时间段参数*/
                        if (taskInfo.getIsFirstExec()){
                                resultDataVo.setStartTime(taskInfo.getStartTime());
                                resultDataVo.setEndTime(taskInfo.getEndTime());
                        }
                        else {
                                long startTime = taskInfo.getStartTime().getTime();
                                long endTime = taskInfo.getEndTime().getTime();
                                Date newStartTime = taskInfo.getEndTime();
                                Date newEndTime = new Date(endTime + (endTime-startTime));
                                resultDataVo.setStartTime(newStartTime);
                                resultDataVo.setEndTime(newEndTime);
                        }
                        HttpResponse getPredictDataRequest = null;
                        try {
                                getPredictDataRequest = HttpUtil.createRequest(Method.POST, predictResultUrl).body(JSON.toJSONString(resultDataVo)).execute();
                        } catch (Exception e) {
                                log.error("调用接口失败：" + e.getMessage());
                                return;
                        }
                        if (null != getPredictDataRequest && getPredictDataRequest.getStatus() == 200){
                                JSONObject result = JSON.parseObject(getPredictDataRequest.body());
                                JSONArray dataList = result.getJSONArray("data");
                                /**查询不到数据就停止任务*/
                                if (null==dataList || dataList.size()==0){
                                        log.error("未查询到对应数据");
                                        return;
                                }else {
                                        Map inputParam =new HashMap<String,Object>();
                                        inputParam.put("dataList",dataList);
                                        String getFaultInfoRequestBody = JSON.toJSONString(inputParam);
                                        /**调用故障模型计算接口*/
                                        HttpResponse getFaultInfoRequest = null;
                                        try {
                                                getFaultInfoRequest = HttpUtil.createRequest(Method.POST, modInfo.getAddr()).body(getFaultInfoRequestBody).execute();
                                        } catch (Exception e) {
                                                log.error("调用接口失败：" + e.getMessage());
                                                return;
                                        }
                                        if (null != getFaultInfoRequest && getFaultInfoRequest.getStatus() == 200){
                                                /**获取返回的故障信息并入库*/
                                                JSONObject faultInfoResult = JSON.parseObject(getFaultInfoRequest.body());
                                                JSONObject faultInfo = faultInfoResult.getJSONObject("data");
                                                FmeaInfo fmeaInfo = JSON.toJavaObject(faultInfo, FmeaInfo.class);
                                                fmeaInfo.setCreateTime(new Date());
                                                fmeaInfo.setStartTime(resultDataVo.getStartTime());
                                                fmeaInfo.setEndTime(resultDataVo.getEndTime());
                                                fmeaInfo.setTaskId(taskInfo.getId());
                                                fmeaInfoDAO.insert(fmeaInfo);

                                                /**构建消息体并发送kafka消息*/
                                                SimpleDateFormat sdf = new SimpleDateFormat(DatePattern.NORM_DATETIME_PATTERN);
                                                PushFaultMsg pushFaultMsg = PushFaultMsg.builder()
                                                        .sendTime(sdf.format(new Date()))
                                                        .failureMode(fmeaInfo.getFailureMode())
                                                        .effect(fmeaInfo.getEffect())
                                                        .severity(fmeaInfo.getSeverity())
                                                        .cause(fmeaInfo.getCause())
                                                        .occurrence(fmeaInfo.getOccurrence())
                                                        .currentProcessControlPlan(fmeaInfo.getCurrentProcessControlPlan())
                                                        .detection(fmeaInfo.getDetection())
                                                        .rpn(fmeaInfo.getRpn())
                                                        .recommendedAction(fmeaInfo.getRecommendedAction())
                                                        .build();
                                                iKafkaProducer.send("FAULT_RESULT",pushFaultMsg);
                                                log.info("发送的kafka信息为============="+ MessageToJsonUtil.toJson(pushFaultMsg));

                                                /**更新任务信息，查询数据时间段变化*/
                                                taskInfo.setStartTime(resultDataVo.getStartTime());
                                                taskInfo.setEndTime(resultDataVo.getEndTime());
                                                taskInfo.setIsFirstExec(false);
                                                updateById(taskInfo);

                                        }
                                        else {
                                                log.error("调用模型计算接口失败");
                                                return;
                                        }
                                }

                        }
                        else{
                                log.error("查询数据失败");
                                return;
                        }
                }
        }

        /**只执行一次计算，不使用定时任务流程，不考虑下次计算*/
        @Override
        @Transactional(rollbackFor = Exception.class)
        public JsonResult<String> computeOnce(FaultTask faultTask){
                FaultTask taskInfo = faultTaskDAO.selectById(faultTask.getId());
                /**查询对应模型信息*/
                List<ModInfo> modInfoList = modInfoDAO.selectByModName(taskInfo.getModelName());
                if (null == modInfoList ){
                        return JsonResult.error("未查询到" +taskInfo.getModelName()+"模型");
                }
                else if (modInfoList.size() != 1){
                        return JsonResult.error("模型名称重复");
                }
                else {
                        /**构建参数，调用查询数据接口*/
                        ModInfo modInfo = modInfoList.get(0);
                        ResultDataVo resultDataVo = ResultDataVo.builder()
                                .satelliteCode(taskInfo.getSatelliteCode())
                                .paramCode(taskInfo.getParamCode())
                                .startTime(taskInfo.getStartTime())
                                .endTime(taskInfo.getEndTime())
                                .build();
                        HttpResponse getPredictDataRequest = null;
                        try {
                                getPredictDataRequest = HttpUtil.createRequest(Method.POST, predictResultUrl).body(JSON.toJSONString(resultDataVo)).execute();
                        } catch (Exception e) {
                                return JsonResult.error("调用接口失败：" + e.getMessage());
                        }
                        if (null != getPredictDataRequest && getPredictDataRequest.getStatus() == 200){
                                JSONObject result = JSON.parseObject(getPredictDataRequest.body());
                                JSONArray dataList = result.getJSONArray("data");
                                /**查询不到数据就停止任务*/
                                if (null==dataList || dataList.size()==0){
                                        return JsonResult.error("未查询到" +taskInfo.getParamCode()+ "数据");
                                }else {
                                        Map inputParam =new HashMap<String,Object>();
                                        inputParam.put("dataList",dataList);
                                        String getFaultInfoRequestBody = JSON.toJSONString(inputParam);
                                        /**调用故障模型计算接口*/
                                        HttpResponse getFaultInfoRequest = null;
                                        try {
                                                getFaultInfoRequest = HttpUtil.createRequest(Method.POST, modInfo.getAddr()).body(getFaultInfoRequestBody).execute();
                                        } catch (Exception e) {
                                                return JsonResult.error("调用接口失败：" + e.getMessage());
                                        }
                                        if (null != getFaultInfoRequest && getFaultInfoRequest.getStatus() == 200){
                                                /**获取返回的故障信息并入库*/
                                                JSONObject faultInfoResult = JSON.parseObject(getFaultInfoRequest.body());
                                                JSONObject faultInfo = faultInfoResult.getJSONObject("data");
                                                FmeaInfo fmeaInfo = JSON.toJavaObject(faultInfo, FmeaInfo.class);
                                                fmeaInfo.setCreateTime(new Date());
                                                fmeaInfo.setStartTime(resultDataVo.getStartTime());
                                                fmeaInfo.setEndTime(resultDataVo.getEndTime());
                                                fmeaInfo.setTaskId(taskInfo.getId());
                                                fmeaInfoDAO.insert(fmeaInfo);

                                                /**构建消息体并发送kafka消息*/
                                                SimpleDateFormat sdf = new SimpleDateFormat(DatePattern.NORM_DATETIME_PATTERN);
                                                PushFaultMsg pushFaultMsg = PushFaultMsg.builder()
                                                        .sendTime(sdf.format(new Date()))
                                                        .failureMode(fmeaInfo.getFailureMode())
                                                        .effect(fmeaInfo.getEffect())
                                                        .severity(fmeaInfo.getSeverity())
                                                        .cause(fmeaInfo.getCause())
                                                        .occurrence(fmeaInfo.getOccurrence())
                                                        .currentProcessControlPlan(fmeaInfo.getCurrentProcessControlPlan())
                                                        .detection(fmeaInfo.getDetection())
                                                        .rpn(fmeaInfo.getRpn())
                                                        .recommendedAction(fmeaInfo.getRecommendedAction())
                                                        .build();
                                                iKafkaProducer.send("FAULT_RESULT",pushFaultMsg);
                                                log.info("发送的kafka信息为============="+ MessageToJsonUtil.toJson(pushFaultMsg));

                                                return JsonResult.ok("调用" +taskInfo.getModelName()+ "模型计算接口成功");
                                        }
                                        else {
                                                return JsonResult.error("调用" +taskInfo.getModelName()+ "模型计算接口失败");
                                        }
                                }
                        }
                        else{
                                return JsonResult.error("查询" +taskInfo.getParamCode()+ "数据接口调用失败");
                        }
                }
        }

        /**检查cron表达式是不是6位*/
        private String checkCron(String cron){
                String newCron = "";
                String[] numstr = cron.split("\\s+");
                if (numstr.length > 6){
                        for (int i=0; i<5 ;i++){
                                newCron += numstr[i] + " ";
                        }
                        newCron += numstr[5];
                        return newCron;
                }
                else
                        return cron;
        }

        @Override
        @Transactional(rollbackFor = Exception.class)
        public JsonResult<String> stopTask(FaultTask faultTask) {
                dynamicTaskService.stopTask(String.valueOf(faultTask.getId()));
                FaultTask taskForDb = faultTaskDAO.selectById(faultTask.getId());
                if (null != taskForDb && taskForDb.getStatus()){
                        taskForDb.setStatus(false);
                        updateById(taskForDb);
                        return JsonResult.ok("任务停止成功: " + faultTask.getTaskName());
                }
                else
                        return JsonResult.error("任务停止失败: " + faultTask.getTaskName());
        }


        @Override
        @Transactional(rollbackFor = Exception.class)
        public FaultTask update(FaultTaskVO vo) {
                FaultTask entity = FaultTaskConvert.INSTANCE.convert(vo);
                updateById(entity);
                return entity;
        }

        @Override
        @Transactional(rollbackFor = Exception.class)
        public void delete(List<Long> idList) {
                removeByIds(idList);
        }

        @Override
        @Transactional(rollbackFor = Exception.class)
        public List<FaultTask> selectByName(String taskName) {
                return faultTaskDAO.selectByName(taskName);
        }

}
